package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.IdUtil;
import com.atguigu.tingshu.common.cache.RedisCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.vo.album.*;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.album.service.AlbumStatService;
import com.atguigu.tingshu.album.service.TrackInfoService;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.TingShuException;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.album.TrackInfo;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	private AlbumAttributeValueService albumAttributeValueService;

	@Autowired
	private AlbumStatService albumStatService;

	@Autowired
	private TrackInfoService trackInfoService;

	@Autowired
	private KafkaService kafkaService;

	@Autowired
	private RedisTemplate redisTemplate;

	@Autowired
	private RedissonClient redissonClient;

	@Transactional(rollbackFor = Exception.class)
	@Override
	public void saveAlbumInfo(AlbumInfoVo albumInfoVo,Long userId) {
		AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo,AlbumInfo.class);
		//模拟数据
		albumInfo.setUserId(1L);
		//试听集数
		albumInfo.setTracksForFree(5);
		//审核通过
		albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
		boolean save = this.save(albumInfo);
		if(!save){
			throw new TingShuException(400,"保存用户信息失败");
		}
		//保存属性
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		if(!CollectionUtils.isEmpty(albumAttributeValueVoList)){
			List<AlbumAttributeValue> albumAttributeValues = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
				AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
				albumAttributeValue.setAlbumId(albumInfo.getId());
				return albumAttributeValue;
			}).collect(Collectors.toList());
			boolean result = albumAttributeValueService.saveBatch(albumAttributeValues);
			if(!result){
				throw new TingShuException(400,"保存专辑属性信息失败");
			}
		}
		//保存专辑统计 统计类型：0401-播放量 0402-订阅量 0403-购买量 0404-评论数
		List<String> strings = List.of(SystemConstant.ALBUM_STAT_PLAY, SystemConstant.ALBUM_STAT_SUBSCRIBE, SystemConstant.ALBUM_STAT_BUY, SystemConstant.ALBUM_STAT_COMMENT);
		List<AlbumStat> albumStats = strings.stream().map(value -> {
			AlbumStat albumStat = new AlbumStat();
			albumStat.setAlbumId(albumInfo.getId());
			albumStat.setStatType(value);
			albumStat.setStatNum(0);
			return albumStat;
		}).collect(Collectors.toList());
		boolean result = albumStatService.saveBatch(albumStats);
		if(!result){
			throw new TingShuException(400,"保存专辑统计信息失败");
		}
		//发送上架信息
		kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER,albumInfo.getId().toString());
		//上架专辑添加到布隆过滤器
		RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
		bloomFilter.add(albumInfo.getId());
	}

	@Override
	public Page<AlbumListVo> findUserAlbumPage(Page<AlbumListVo> pageInfo, AlbumInfoQuery albumInfoQuery) {
		return baseMapper.findUserAlbumPage(pageInfo,albumInfoQuery);
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public void removeAlbumInfo(Long id) {
		//判断专辑是否存在关联声音
		long count = trackInfoService.count(new LambdaQueryWrapper<TrackInfo>().eq(TrackInfo::getAlbumId, id));
		if(count > 0){
			throw new TingShuException(400,"该专辑下存在未删除的声音");
		}
		//删除专辑关联属性
		boolean remove = albumAttributeValueService.remove(new LambdaQueryWrapper<AlbumAttributeValue>().in(AlbumAttributeValue::getAlbumId, id));
		//删除专辑统计信息
		boolean removeStat = albumStatService.remove(new LambdaQueryWrapper<AlbumStat>().in(AlbumStat::getAlbumId, id));
		//删除专辑信息
		boolean result = this.removeById(id);
		if(!remove){
			throw new TingShuException(400,"删除专辑失败");
		}
		//发送下架信息
		kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER,id.toString());
	}

	@Override
	@RedisCache(prefix =
			RedisConstant.ALBUM_INFO_PREFIX,
			tempTimeOut = RedisConstant.ALBUM_TEMPORARY_TIMEOUT ,
			timeOut = RedisConstant.ALBUM_TIMEOUT,
			timeUnit = TimeUnit.SECONDS)
	public AlbumInfo getAlbumInfo(Long id) {
		return getAlbumInfoByDB(id);
	}

	private AlbumInfo getAlbumInfoByRedisson(Long id) {
		try {
			String key = RedisConstant.ALBUM_INFO_PREFIX + id;
			AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(key);
			if(albumInfo != null){
				return albumInfo;
			}
			//使用redisson分布式锁
			String lockKey = RedisConstant.ALBUM_LOCK_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
			RLock lock = redissonClient.getLock(lockKey);
			try {
				lock.lock();
				//二次校验获取
				albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(key);
				if(albumInfo != null){
					return albumInfo;
				}
				AlbumInfo albumInfoByDB = this.getAlbumInfoByDB(id);
				if(albumInfoByDB == null) {
					albumInfoByDB = new AlbumInfo();
					redisTemplate.opsForValue().set(key, albumInfoByDB, RedisConstant.ALBUM_TEMPORARY_TIMEOUT,TimeUnit.SECONDS);
					return albumInfoByDB;
				}else{
					redisTemplate.opsForValue().set(key, albumInfoByDB, RedisConstant.ALBUM_TIMEOUT,TimeUnit.SECONDS);
					return albumInfoByDB;
				}
			} finally {
				lock.unlock();
			}
		} catch (Exception e) {
			e.printStackTrace();
			//数据库兜底方法
			return getAlbumInfoByRedis(id);
		}
	}

	/**
	 * 通过redis获取专辑信息
	 * @param id
	 * @return
	 */
	private @Nullable AlbumInfo getAlbumInfoByRedis(Long id) {
		try {
			String key = RedisConstant.ALBUM_INFO_PREFIX + id;
			AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(key);
			if (albumInfo == null) {
				//尝试获取分布式锁
				String lockValue = IdUtil.fastSimpleUUID();
				String lockKey = RedisConstant.ALBUM_LOCK_PREFIX + id;
				Boolean b = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
				if (b) {
					try {
						AlbumInfo albumInfoByDB = getAlbumInfoByDB(id);
						if (albumInfoByDB == null) {
							albumInfoByDB = new AlbumInfo();
							redisTemplate.opsForValue().set(key, albumInfoByDB, RedisConstant.ALBUM_TEMPORARY_TIMEOUT,TimeUnit.SECONDS);
							return albumInfoByDB;
						} else {
							redisTemplate.opsForValue().set(key, albumInfoByDB, RedisConstant.ALBUM_TIMEOUT,TimeUnit.SECONDS);
							return albumInfoByDB;
						}
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						//lua脚本释放锁
						DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
						//4.2设置脚本文本
						String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
								"then\n" +
								"    return redis.call(\"del\",KEYS[1])\n" +
								"else\n" +
								"    return 0\n" +
								"end";
						redisScript.setScriptText(script);
						//4.3 设置响应类型
						redisScript.setResultType(Long.class);
						redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockValue);
					}
				} else {
					try {
						Thread.sleep(1000);
						return getAlbumInfo(id);
					} catch (InterruptedException e) {
						throw new RuntimeException(e);
					}
				}

			}
			return albumInfo;
		} catch (RuntimeException e) {
			e.printStackTrace();
		}
		//兜底方法
		return getAlbumInfoByDB(id);
	}

	/**
	 * 数据库查询
	 * @param id
	 * @return
	 */
	private AlbumInfo getAlbumInfoByDB(Long id) {
		AlbumInfo albumInfo = this.getById(id);
		if(albumInfo == null){
			return null;
		}
		List<AlbumAttributeValue> list = albumAttributeValueService
					    .list(new LambdaQueryWrapper<AlbumAttributeValue>()
						.eq(AlbumAttributeValue::getAlbumId, id)
								.eq(AlbumAttributeValue::getIsDeleted,0));
		albumInfo.setAlbumAttributeValueVoList(list);
		return albumInfo;
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public void updateAlbumInfo(Long id,AlbumInfoVo albumInfoVo) {
		AlbumInfo albumInfoDB = this.getById(id);
		if(albumInfoDB == null){
			throw new TingShuException(400,"专辑信息不存在");
		}
		//删除原来的属性关联
		albumAttributeValueService.remove(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, id));
		BeanUtils.copyProperties(albumInfoVo,albumInfoDB);
		albumInfoDB.setUpdateTime(new Date());
		boolean result = this.updateById(albumInfoDB);
		if(!result){
			throw new TingShuException(400,"修改专辑信息失败");
		}
		//保存属性
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		if(!CollectionUtils.isEmpty(albumAttributeValueVoList)){
			List<AlbumAttributeValue> albumAttributeValues = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
				AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
				albumAttributeValue.setAlbumId(id);
				return albumAttributeValue;
			}).collect(Collectors.toList());
			boolean resultSave = albumAttributeValueService.saveBatch(albumAttributeValues);
			if(!resultSave){
				throw new TingShuException(400,"保存专辑属性信息失败");
			}
		}

		if(albumInfoVo.getIsOpen().equals("0")){
			kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER,id.toString());
		}else{
			kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER,id.toString());
		}
	}

	@Override
	public List<AlbumInfo> findUserAllAlbumList(Long userId) {
		return this.list(new LambdaQueryWrapper<AlbumInfo>()
				.eq(AlbumInfo::getUserId,userId)
				.eq(AlbumInfo::getIsDeleted,0)
				.orderByAsc(AlbumInfo::getId)
				.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle)
				.last("limit 100"));
	}

	/**
	 * 获取专辑统计信息
	 * @param albumId
	 * @return
	 */
	@Override
	@RedisCache(prefix = RedisConstant.ALBUM_STAT_PREFIX)
	public AlbumStatVo getAlbumStatVo(Long albumId) {
		return baseMapper.getAlbumStatVo(albumId);
	}

}
